package com.zhang.gmall.realtime.app.dwm;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONAware;
import com.alibaba.fastjson.JSONObject;
import com.zhang.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import java.text.SimpleDateFormat;

/**
 * @title: 独立访客计算 uv
 * @author: zhang
 * @date: 2022/3/6 14:24
 */
public class UniqueVisitApp {
    public static void main(String[] args) throws Exception {
        // TODO 1.获取环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        // todo 检查点设置
        // TODO 2.从kafka读取数据 、对数据进行结构转换、对页面日志按mid进行分组、在一天时间范围内计算uv 只保存每个mid每天第一次登录的数据
        String topic = "dwd_page_log_2022";
        String sinkTopic = "dwm_unique_visit_2022";
        String groupId = "dwm_uv_app";
        SingleOutputStreamOperator<JSONObject> resultDS = env
                .addSource(MyKafkaUtil.getKafkaSource(topic, groupId))
                .map(JSON::parseObject)
                .keyBy(jsonObj -> jsonObj.getJSONObject("common").getString("mid"))
                .process(new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                    private ValueState<String> lastViewTime;
                    private SimpleDateFormat sdf;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        //设置状态过期时间
                        ValueStateDescriptor<String> stateDescriptor = new ValueStateDescriptor<>("lastViewTime", Types.STRING);
                        StateTtlConfig stateTtlConfig = new StateTtlConfig
                                .Builder(Time.hours(24))
                                //当状态更新和重写时候 重新倒计时
                                .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                                .build();
                        stateDescriptor.enableTimeToLive(stateTtlConfig);
                        lastViewTime = getRuntimeContext().getState(stateDescriptor);

                        sdf = new SimpleDateFormat("yyyy-MM-dd");
                    }

                    @Override
                    public void processElement(JSONObject jsonObj, KeyedProcessFunction<String, JSONObject, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                        String lastPage = jsonObj.getJSONObject("page").getString("last_page_id");
                        //如果上次访问页面存在，则一定不是第一次访问，直接丢弃
                        if (lastPage == null || lastPage.length() <= 0) {
                            String lastTime = lastViewTime.value();
                            long ts = jsonObj.getLong("ts");
                            String curTime = sdf.format(ts);
                            //如果上次访问日期为空，或者不是今天说明今天没有访问过
                            if (lastTime == null || !lastTime.equals(curTime)) {
                                lastViewTime.update(curTime);
                                out.collect(jsonObj);
                            }
                        }
                    }
                });

        // TODO 3.写入kafka dwm层主题
        resultDS.print("uv");
        resultDS.map(JSONAware::toJSONString).addSink(MyKafkaUtil.getKafkaSink(sinkTopic));
        // TODO 4.执行任务
        env.execute();
    }
}
